package com.atguigu.outputformat;

import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

public class MyRecordWriter extends RecordWriter<LongWritable, Text> {

    private FSDataOutputStream atguigu;
    private FSDataOutputStream others;

    public void initialize(TaskAttemptContext job) throws IOException {
        // 从配置文件中 获取输出路径
        String outDir = job.getConfiguration().get(FileOutputFormat.OUTDIR);
        // 获取本次job的文件系统
        FileSystem fileSystem = FileSystem.get(job.getConfiguration());
        // 使用本次文件系统 开流
        atguigu = fileSystem.create(new Path(outDir + "//atguigu.log"));
        others = fileSystem.create(new Path(outDir + "//others.log"));
    }

    /**
     * 将KV写出，没对KV调用一次
     *
     * @param key
     * @param value
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    public void write(LongWritable key, Text value) throws IOException, InterruptedException {
        String out = value.toString() + "\n";
        if (out.contains("atguigu")){
            atguigu.write(out.getBytes());
        }else {
            others.write(out.getBytes());
        }
    }

    /**
     * 关闭资源
     *
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    public void close(TaskAttemptContext context) throws IOException, InterruptedException {
        IOUtils.closeStream(atguigu);
        IOUtils.closeStream(others);
    }
}
